package com.wtz.kafka;

import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.wtz.kafka.entity.RandomValue;
import org.apache.commons.math3.random.RandomDataGenerator;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.math.BigDecimal;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.util.Properties;
import java.util.UUID;

/**
 * @author : tiezhu
 * @date : 2020/3/25
 */
public class KafkaProducerFlinkStreamSql {

    private static final String BROKER_LIST = "kudu1:9092";

    public static void main(String[] args) throws JSONException, InterruptedException {

        final String TOPIC = args[0];

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        int i = 1;
        while (true) {
            sendData(producer, TOPIC, RandomValue.buildJsonData(i++));
        }
    }

    private static void sendData(KafkaProducer<String, String> producer, String topic, String data) throws InterruptedException {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, data);
        System.out.println(data);
        producer.send(record);
//        Thread.sleep(500);
    }
}
